/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server;

import java.io.BufferedWriter;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.Writer;
import java.lang.management.ManagementFactory;
import java.lang.management.OperatingSystemMXBean;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.jute.BinaryInputArchive;
import org.apache.jute.BinaryOutputArchive;
import org.apache.jute.Record;
import org.apache.zookeeper.Environment;
import org.apache.zookeeper.Version;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.proto.ReplyHeader;
import org.apache.zookeeper.proto.RequestHeader;
import org.apache.zookeeper.proto.WatcherEvent;
import org.apache.zookeeper.server.quorum.Leader;
import org.apache.zookeeper.server.quorum.LeaderZooKeeperServer;
import org.apache.zookeeper.server.quorum.ReadOnlyZooKeeperServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.sun.management.UnixOperatingSystemMXBean;

/**
 * This class handles communication with clients using NIO. There is one per
 * client, but only one thread doing the communication.
 */
public class NIOServerCnxn extends ServerCnxn {
	private class CnxnStatResetCommand extends CommandThread {
		public CnxnStatResetCommand(PrintWriter pw) {
			super(pw);
		}

		@Override
		public void commandRun() {
			if (NIOServerCnxn.this.zkServer == null) {
				this.pw.println(ZK_NOT_SERVING);
			} else {
				synchronized (NIOServerCnxn.this.factory.cnxns) {
					for (ServerCnxn c : NIOServerCnxn.this.factory.cnxns) {
						c.resetStats();
					}
				}
				this.pw.println("Connection stats reset.");
			}
		}
	}

	/**
	 * Set of threads for commmand ports. All the 4 letter commands are run via a
	 * thread. Each class maps to a corresponding 4 letter command. CommandThread is
	 * the abstract class from which all the others inherit.
	 */
	private abstract class CommandThread extends Thread {
		PrintWriter pw;

		CommandThread(PrintWriter pw) {
			this.pw = pw;
		}

		public abstract void commandRun() throws IOException;

		@Override
		public void run() {
			try {
				this.commandRun();
			} catch (IOException ie) {
				LOG.error("Error in running command ", ie);
			} finally {
				NIOServerCnxn.this.cleanupWriterSocket(this.pw);
			}
		}
	}

	private class ConfCommand extends CommandThread {
		ConfCommand(PrintWriter pw) {
			super(pw);
		}

		@Override
		public void commandRun() {
			if (NIOServerCnxn.this.zkServer == null) {
				this.pw.println(ZK_NOT_SERVING);
			} else {
				NIOServerCnxn.this.zkServer.dumpConf(this.pw);
			}
		}
	}

	private class ConsCommand extends CommandThread {
		public ConsCommand(PrintWriter pw) {
			super(pw);
		}

		@SuppressWarnings("unchecked")
		@Override
		public void commandRun() {
			if (NIOServerCnxn.this.zkServer == null) {
				this.pw.println(ZK_NOT_SERVING);
			} else {
				// clone should be faster than iteration
				// ie give up the cnxns lock faster
				HashSet<NIOServerCnxn> cnxns;
				synchronized (NIOServerCnxn.this.factory.cnxns) {
					cnxns = (HashSet<NIOServerCnxn>) NIOServerCnxn.this.factory.cnxns.clone();
				}
				for (NIOServerCnxn c : cnxns) {
					c.dumpConnectionInfo(this.pw, false);
					this.pw.println();
				}
				this.pw.println();
			}
		}
	}

	private class DumpCommand extends CommandThread {
		public DumpCommand(PrintWriter pw) {
			super(pw);
		}

		@Override
		public void commandRun() {
			if (NIOServerCnxn.this.zkServer == null) {
				this.pw.println(ZK_NOT_SERVING);
			} else {
				this.pw.println("SessionTracker dump:");
				NIOServerCnxn.this.zkServer.sessionTracker.dumpSessions(this.pw);
				this.pw.println("ephemeral nodes dump:");
				NIOServerCnxn.this.zkServer.dumpEphemerals(this.pw);
			}
		}
	}

	private class EnvCommand extends CommandThread {
		EnvCommand(PrintWriter pw) {
			super(pw);
		}

		@Override
		public void commandRun() {
			List<Environment.Entry> env = Environment.list();

			this.pw.println("Environment:");
			for (Environment.Entry e : env) {
				this.pw.print(e.getKey());
				this.pw.print("=");
				this.pw.println(e.getValue());
			}

		}
	}

	private class IsroCommand extends CommandThread {

		public IsroCommand(PrintWriter pw) {
			super(pw);
		}

		@Override
		public void commandRun() {
			if (NIOServerCnxn.this.zkServer == null) {
				this.pw.print("null");
			} else if (NIOServerCnxn.this.zkServer instanceof ReadOnlyZooKeeperServer) {
				this.pw.print("ro");
			} else {
				this.pw.print("rw");
			}
		}
	}

	private class MonitorCommand extends CommandThread {

		MonitorCommand(PrintWriter pw) {
			super(pw);
		}

		@Override
		public void commandRun() {
			if (NIOServerCnxn.this.zkServer == null) {
				this.pw.println(ZK_NOT_SERVING);
				return;
			}
			ZKDatabase zkdb = NIOServerCnxn.this.zkServer.getZKDatabase();
			ServerStats stats = NIOServerCnxn.this.zkServer.serverStats();

			this.print("version", Version.getFullVersion());

			this.print("avg_latency", stats.getAvgLatency());
			this.print("max_latency", stats.getMaxLatency());
			this.print("min_latency", stats.getMinLatency());

			this.print("packets_received", stats.getPacketsReceived());
			this.print("packets_sent", stats.getPacketsSent());

			this.print("outstanding_requests", stats.getOutstandingRequests());

			this.print("server_state", stats.getServerState());
			this.print("znode_count", zkdb.getNodeCount());

			this.print("watch_count", zkdb.getDataTree().getWatchCount());
			this.print("ephemerals_count", zkdb.getDataTree().getEphemeralsCount());
			this.print("approximate_data_size", zkdb.getDataTree().approximateDataSize());

			OperatingSystemMXBean osMbean = ManagementFactory.getOperatingSystemMXBean();
			if (osMbean != null && osMbean instanceof UnixOperatingSystemMXBean) {
				UnixOperatingSystemMXBean unixos = (UnixOperatingSystemMXBean) osMbean;

				this.print("open_file_descriptor_count", unixos.getOpenFileDescriptorCount());
				this.print("max_file_descriptor_count", unixos.getMaxFileDescriptorCount());
			}

			if (stats.getServerState().equals("leader")) {
				Leader leader = ((LeaderZooKeeperServer) NIOServerCnxn.this.zkServer).getLeader();

				this.print("followers", leader.learners.size());
				this.print("synced_followers", leader.forwardingFollowers.size());
				this.print("pending_syncs", leader.pendingSyncs.size());
			}
		}

		private void print(String key, long number) {
			this.print(key, "" + number);
		}

		private void print(String key, String value) {
			this.pw.print("zk_");
			this.pw.print(key);
			this.pw.print("\t");
			this.pw.println(value);
		}

	}

	private class RuokCommand extends CommandThread {
		public RuokCommand(PrintWriter pw) {
			super(pw);
		}

		@Override
		public void commandRun() {
			this.pw.print("imok");

		}
	}

	/**
	 * This class wraps the sendBuffer method of NIOServerCnxn. It is responsible
	 * for chunking up the response to a client. Rather than cons'ing up a response
	 * fully in memory, which may be large for some commands, this class chunks up
	 * the result.
	 */
	private class SendBufferWriter extends Writer {
		private StringBuffer sb = new StringBuffer();

		/**
		 * Check if we are ready to send another chunk.
		 * 
		 * @param force force sending, even if not a full chunk
		 */
		private void checkFlush(boolean force) {
			if ((force && this.sb.length() > 0) || this.sb.length() > 2048) {
				NIOServerCnxn.this.sendBufferSync(ByteBuffer.wrap(this.sb.toString().getBytes()));
				// clear our internal buffer
				this.sb.setLength(0);
			}
		}

		@Override
		public void close() throws IOException {
			if (this.sb == null)
				return;
			this.checkFlush(true);
			this.sb = null; // clear out the ref to ensure no reuse
		}

		@Override
		public void flush() throws IOException {
			this.checkFlush(true);
		}

		@Override
		public void write(char[] cbuf, int off, int len) throws IOException {
			this.sb.append(cbuf, off, len);
			this.checkFlush(false);
		}
	}

	private class SetTraceMaskCommand extends CommandThread {
		long trace = 0;

		SetTraceMaskCommand(PrintWriter pw, long trace) {
			super(pw);
			this.trace = trace;
		}

		@Override
		public void commandRun() {
			this.pw.print(this.trace);
		}
	}

	private class StatCommand extends CommandThread {
		int len;

		public StatCommand(PrintWriter pw, int len) {
			super(pw);
			this.len = len;
		}

		@SuppressWarnings("unchecked")
		@Override
		public void commandRun() {
			if (NIOServerCnxn.this.zkServer == null) {
				this.pw.println(ZK_NOT_SERVING);
			} else {
				this.pw.print("Zookeeper version: ");
				this.pw.println(Version.getFullVersion());
				if (NIOServerCnxn.this.zkServer instanceof ReadOnlyZooKeeperServer) {
					this.pw.println("READ-ONLY mode; serving only " + "read-only clients");
				}
				if (this.len == statCmd) {
					LOG.info("Stat command output");
					this.pw.println("Clients:");
					// clone should be faster than iteration
					// ie give up the cnxns lock faster
					HashSet<NIOServerCnxn> cnxnset;
					synchronized (NIOServerCnxn.this.factory.cnxns) {
						cnxnset = (HashSet<NIOServerCnxn>) NIOServerCnxn.this.factory.cnxns.clone();
					}
					for (NIOServerCnxn c : cnxnset) {
						c.dumpConnectionInfo(this.pw, true);
						this.pw.println();
					}
					this.pw.println();
				}
				this.pw.print(NIOServerCnxn.this.zkServer.serverStats().toString());
				this.pw.print("Node count: ");
				this.pw.println(NIOServerCnxn.this.zkServer.getZKDatabase().getNodeCount());
			}

		}
	}

	private class StatResetCommand extends CommandThread {
		public StatResetCommand(PrintWriter pw) {
			super(pw);
		}

		@Override
		public void commandRun() {
			if (NIOServerCnxn.this.zkServer == null) {
				this.pw.println(ZK_NOT_SERVING);
			} else {
				NIOServerCnxn.this.zkServer.serverStats().reset();
				this.pw.println("Server stats reset.");
			}
		}
	}

	private class TraceMaskCommand extends CommandThread {
		TraceMaskCommand(PrintWriter pw) {
			super(pw);
		}

		@Override
		public void commandRun() {
			long traceMask = ZooTrace.getTextTraceLevel();
			this.pw.print(traceMask);
		}
	}

	private class WatchCommand extends CommandThread {
		int len = 0;

		public WatchCommand(PrintWriter pw, int len) {
			super(pw);
			this.len = len;
		}

		@Override
		public void commandRun() {
			if (NIOServerCnxn.this.zkServer == null) {
				this.pw.println(ZK_NOT_SERVING);
			} else {
				DataTree dt = NIOServerCnxn.this.zkServer.getZKDatabase().getDataTree();
				if (this.len == wchsCmd) {
					dt.dumpWatchesSummary(this.pw);
				} else if (this.len == wchpCmd) {
					dt.dumpWatches(this.pw, true);
				} else {
					dt.dumpWatches(this.pw, false);
				}
				this.pw.println();
			}
		}
	}

	private final static byte fourBytes[] = new byte[4];

	static final Logger LOG = LoggerFactory.getLogger(NIOServerCnxn.class);

	static long nextSessionId = 1;

	private static final String ZK_NOT_SERVING = "This ZooKeeper instance is not currently serving requests";

	NIOServerCnxnFactory factory;

	ByteBuffer lenBuffer = ByteBuffer.allocate(4);

	ByteBuffer incomingBuffer = this.lenBuffer;

	boolean initialized;

	LinkedBlockingQueue<ByteBuffer> outgoingBuffers = new LinkedBlockingQueue<ByteBuffer>();

	int outstandingLimit = 1;

	/**
	 * The number of requests that have been submitted but not yet responded to.
	 */
	int outstandingRequests;

	/**
	 * This is the id that uniquely identifies the session of a client. Once this
	 * session is no longer active, the ephemeral nodes will go away.
	 */
	long sessionId;

	int sessionTimeout;

	private SelectionKey sk;

	SocketChannel sock;

	private final ZooKeeperServer zkServer;

	public NIOServerCnxn(ZooKeeperServer zk, SocketChannel sock, SelectionKey sk, NIOServerCnxnFactory factory)
			throws IOException {
		this.zkServer = zk;
		this.sock = sock;
		this.sk = sk;
		this.factory = factory;
		if (this.factory.login != null) {
			this.zooKeeperSaslServer = new ZooKeeperSaslServer(factory.login);
		}
		if (zk != null) {
			this.outstandingLimit = zk.getGlobalOutstandingLimit();
		}
		sock.socket().setTcpNoDelay(true);
		/*
		 * set socket linger to false, so that socket close does not block
		 */
		sock.socket().setSoLinger(false, -1);
		InetAddress addr = ((InetSocketAddress) sock.socket().getRemoteSocketAddress()).getAddress();
		this.authInfo.add(new Id("ip", addr.getHostAddress()));
		sk.interestOps(SelectionKey.OP_READ);
	}

	/** Return if four letter word found and responded to, otw false **/
	private boolean checkFourLetterWord(final SelectionKey k, final int len) throws IOException {
		// We take advantage of the limited size of the length to look
		// for cmds. They are all 4-bytes which fits inside of an int
		String cmd = cmd2String.get(len);
		if (cmd == null) {
			return false;
		}
		LOG.info("Processing " + cmd + " command from " + this.sock.socket().getRemoteSocketAddress());
		this.packetReceived();

		/**
		 * cancel the selection key to remove the socket handling from selector. This is
		 * to prevent netcat problem wherein netcat immediately closes the sending side
		 * after sending the commands and still keeps the receiving channel open. The
		 * idea is to remove the selectionkey from the selector so that the selector
		 * does not notice the closed read on the socket channel and keep the socket
		 * alive to write the data to and makes sure to close the socket after its done
		 * writing the data
		 */
		if (k != null) {
			try {
				k.cancel();
			} catch (Exception e) {
				LOG.error("Error cancelling command selection key ", e);
			}
		}

		final PrintWriter pwriter = new PrintWriter(new BufferedWriter(new SendBufferWriter()));
		if (len == ruokCmd) {
			RuokCommand ruok = new RuokCommand(pwriter);
			ruok.start();
			return true;
		} else if (len == getTraceMaskCmd) {
			TraceMaskCommand tmask = new TraceMaskCommand(pwriter);
			tmask.start();
			return true;
		} else if (len == setTraceMaskCmd) {
			int rc = this.sock.read(this.incomingBuffer);
			if (rc < 0) {
				throw new IOException("Read error");
			}

			this.incomingBuffer.flip();
			long traceMask = this.incomingBuffer.getLong();
			ZooTrace.setTextTraceLevel(traceMask);
			SetTraceMaskCommand setMask = new SetTraceMaskCommand(pwriter, traceMask);
			setMask.start();
			return true;
		} else if (len == enviCmd) {
			EnvCommand env = new EnvCommand(pwriter);
			env.start();
			return true;
		} else if (len == confCmd) {
			ConfCommand ccmd = new ConfCommand(pwriter);
			ccmd.start();
			return true;
		} else if (len == srstCmd) {
			StatResetCommand strst = new StatResetCommand(pwriter);
			strst.start();
			return true;
		} else if (len == crstCmd) {
			CnxnStatResetCommand crst = new CnxnStatResetCommand(pwriter);
			crst.start();
			return true;
		} else if (len == dumpCmd) {
			DumpCommand dump = new DumpCommand(pwriter);
			dump.start();
			return true;
		} else if (len == statCmd || len == srvrCmd) {
			StatCommand stat = new StatCommand(pwriter, len);
			stat.start();
			return true;
		} else if (len == consCmd) {
			ConsCommand cons = new ConsCommand(pwriter);
			cons.start();
			return true;
		} else if (len == wchpCmd || len == wchcCmd || len == wchsCmd) {
			WatchCommand wcmd = new WatchCommand(pwriter, len);
			wcmd.start();
			return true;
		} else if (len == mntrCmd) {
			MonitorCommand mntr = new MonitorCommand(pwriter);
			mntr.start();
			return true;
		} else if (len == isroCmd) {
			IsroCommand isro = new IsroCommand(pwriter);
			isro.start();
			return true;
		}
		return false;
	}

	/**
	 * clean up the socket related to a command and also make sure we flush the data
	 * before we do that
	 * 
	 * @param pwriter the pwriter for a command socket
	 */
	private void cleanupWriterSocket(PrintWriter pwriter) {
		try {
			if (pwriter != null) {
				pwriter.flush();
				pwriter.close();
			}
		} catch (Exception e) {
			LOG.info("Error closing PrintWriter ", e);
		} finally {
			try {
				this.close();
			} catch (Exception e) {
				LOG.error("Error closing a command socket ", e);
			}
		}
	}

	/*
	 * Close the cnxn and remove it from the factory cnxns list.
	 * 
	 * This function returns immediately if the cnxn is not on the cnxns list.
	 */
	@Override
	public void close() {
		synchronized (this.factory.cnxns) {
			// if this is not in cnxns then it's already closed
			if (!this.factory.cnxns.remove(this)) {
				return;
			}

			synchronized (this.factory.ipMap) {
				Set<NIOServerCnxn> s = this.factory.ipMap.get(this.sock.socket().getInetAddress());
				s.remove(this);
			}

			this.factory.unregisterConnection(this);

			if (this.zkServer != null) {
				this.zkServer.removeCnxn(this);
			}

			this.closeSock();

			if (this.sk != null) {
				try {
					// need to cancel this selection key from the selector
					this.sk.cancel();
				} catch (Exception e) {
					if (LOG.isDebugEnabled()) {
						LOG.debug("ignoring exception during selectionkey cancel", e);
					}
				}
			}
		}
	}

	/**
	 * Close resources associated with the sock of this cnxn.
	 */
	private void closeSock() {
		if (this.sock == null) {
			return;
		}

		LOG.info("Closed socket connection for client " + this.sock.socket().getRemoteSocketAddress()
				+ (this.sessionId != 0 ? " which had sessionid 0x" + Long.toHexString(this.sessionId)
						: " (no session established for client)"));
		try {
			/*
			 * The following sequence of code is stupid! You would think that only
			 * sock.close() is needed, but alas, it doesn't work that way. If you just do
			 * sock.close() there are cases where the socket doesn't actually close...
			 */
			this.sock.socket().shutdownOutput();
		} catch (IOException e) {
			// This is a relatively common exception that we can't avoid
			if (LOG.isDebugEnabled()) {
				LOG.debug("ignoring exception during output shutdown", e);
			}
		}
		try {
			this.sock.socket().shutdownInput();
		} catch (IOException e) {
			// This is a relatively common exception that we can't avoid
			if (LOG.isDebugEnabled()) {
				LOG.debug("ignoring exception during input shutdown", e);
			}
		}
		try {
			this.sock.socket().close();
		} catch (IOException e) {
			if (LOG.isDebugEnabled()) {
				LOG.debug("ignoring exception during socket close", e);
			}
		}
		try {
			this.sock.close();
			// XXX The next line doesn't seem to be needed, but some posts
			// to forums suggest that it is needed. Keep in mind if errors in
			// this section arise.
			// factory.selector.wakeup();
		} catch (IOException e) {
			if (LOG.isDebugEnabled()) {
				LOG.debug("ignoring exception during socketchannel close", e);
			}
		}
		this.sock = null;
	}

	@Override
	public void disableRecv() {
		this.sk.interestOps(this.sk.interestOps() & (~SelectionKey.OP_READ));
	}

	void doIO(SelectionKey k) throws InterruptedException {
		try {
			if (this.sock == null) {
				LOG.warn("trying to do i/o on a null socket for session:0x" + Long.toHexString(this.sessionId));

				return;
			}
			if (k.isReadable()) {
				int rc = this.sock.read(this.incomingBuffer);
				if (rc < 0) {
					throw new EndOfStreamException("Unable to read additional data from client sessionid 0x"
							+ Long.toHexString(this.sessionId) + ", likely client has closed socket");
				}
				if (this.incomingBuffer.remaining() == 0) {
					boolean isPayload;
					if (this.incomingBuffer == this.lenBuffer) { // start of next request
						this.incomingBuffer.flip();
						isPayload = this.readLength(k);
						this.incomingBuffer.clear();
					} else {
						// continuation
						isPayload = true;
					}
					if (isPayload) { // not the case for 4letterword
						this.readPayload();
					} else {
						// four letter words take care
						// need not do anything else
						return;
					}
				}
			}
			if (k.isWritable()) {
				// ZooLog.logTraceMessage(LOG,
				// ZooLog.CLIENT_DATA_PACKET_TRACE_MASK
				// "outgoingBuffers.size() = " +
				// outgoingBuffers.size());
				if (this.outgoingBuffers.size() > 0) {
					// ZooLog.logTraceMessage(LOG,
					// ZooLog.CLIENT_DATA_PACKET_TRACE_MASK,
					// "sk " + k + " is valid: " +
					// k.isValid());

					/*
					 * This is going to reset the buffer position to 0 and the limit to the size of
					 * the buffer, so that we can fill it with data from the non-direct buffers that
					 * we need to send.
					 */
					ByteBuffer directBuffer = this.factory.directBuffer;
					directBuffer.clear();

					for (ByteBuffer b : this.outgoingBuffers) {
						if (directBuffer.remaining() < b.remaining()) {
							/*
							 * When we call put later, if the directBuffer is to small to hold everything,
							 * nothing will be copied, so we've got to slice the buffer if it's too big.
							 */
							b = (ByteBuffer) b.slice().limit(directBuffer.remaining());
						}
						/*
						 * put() is going to modify the positions of both buffers, put we don't want to
						 * change the position of the source buffers (we'll do that after the send, if
						 * needed), so we save and reset the position after the copy
						 */
						int p = b.position();
						directBuffer.put(b);
						b.position(p);
						if (directBuffer.remaining() == 0) {
							break;
						}
					}
					/*
					 * Do the flip: limit becomes position, position gets set to 0. This sets us up
					 * for the write.
					 */
					directBuffer.flip();

					int sent = this.sock.write(directBuffer);
					ByteBuffer bb;

					// Remove the buffers that we have sent
					while (this.outgoingBuffers.size() > 0) {
						bb = this.outgoingBuffers.peek();
						if (bb == ServerCnxnFactory.closeConn) {
							throw new CloseRequestException("close requested");
						}
						int left = bb.remaining() - sent;
						if (left > 0) {
							/*
							 * We only partially sent this buffer, so we update the position and exit the
							 * loop.
							 */
							bb.position(bb.position() + sent);
							break;
						}
						this.packetSent();
						/* We've sent the whole buffer, so drop the buffer */
						sent -= bb.remaining();
						this.outgoingBuffers.remove();
					}
					// ZooLog.logTraceMessage(LOG,
					// ZooLog.CLIENT_DATA_PACKET_TRACE_MASK, "after send,
					// outgoingBuffers.size() = " + outgoingBuffers.size());
				}

				synchronized (this.factory) {
					if (this.outgoingBuffers.size() == 0) {
						if (!this.initialized && (this.sk.interestOps() & SelectionKey.OP_READ) == 0) {
							throw new CloseRequestException("responded to info probe");
						}
						this.sk.interestOps(this.sk.interestOps() & (~SelectionKey.OP_WRITE));
					} else {
						this.sk.interestOps(this.sk.interestOps() | SelectionKey.OP_WRITE);
					}
				}
			}
		} catch (CancelledKeyException e) {
			LOG.warn("Exception causing close of session 0x" + Long.toHexString(this.sessionId) + " due to " + e);
			if (LOG.isDebugEnabled()) {
				LOG.debug("CancelledKeyException stack trace", e);
			}
			this.close();
		} catch (CloseRequestException e) {
			// expecting close to log session closure
			this.close();
		} catch (EndOfStreamException e) {
			LOG.warn("caught end of stream exception", e); // tell user why

			// expecting close to log session closure
			this.close();
		} catch (IOException e) {
			LOG.warn("Exception causing close of session 0x" + Long.toHexString(this.sessionId) + " due to " + e);
			if (LOG.isDebugEnabled()) {
				LOG.debug("IOException stack trace", e);
			}
			this.close();
		}
	}

	@Override
	public void enableRecv() {
		synchronized (this.factory) {
			this.sk.selector().wakeup();
			if (this.sk.isValid()) {
				int interest = this.sk.interestOps();
				if ((interest & SelectionKey.OP_READ) == 0) {
					this.sk.interestOps(interest | SelectionKey.OP_READ);
				}
			}
		}
	}

	@Override
	public int getInterestOps() {
		return this.sk.isValid() ? this.sk.interestOps() : 0;
	}

	@Override
	public long getOutstandingRequests() {
		synchronized (this) {
			synchronized (this.factory) {
				return this.outstandingRequests;
			}
		}
	}

	@Override
	public InetSocketAddress getRemoteSocketAddress() {
		if (this.sock == null) {
			return null;
		}
		return (InetSocketAddress) this.sock.socket().getRemoteSocketAddress();
	}

	/*
	 * (non-Javadoc)
	 *
	 * @see org.apache.zookeeper.server.ServerCnxnIface#getSessionId()
	 */
	@Override
	public long getSessionId() {
		return this.sessionId;
	}

	/*
	 * (non-Javadoc)
	 *
	 * @see org.apache.zookeeper.server.ServerCnxnIface#getSessionTimeout()
	 */
	@Override
	public int getSessionTimeout() {
		return this.sessionTimeout;
	}

	@Override
	protected void incrOutstandingRequests(RequestHeader h) {
		if (h.getXid() >= 0) {
			synchronized (this) {
				this.outstandingRequests++;
			}
			synchronized (this.factory) {
				// check throttling
				if (this.zkServer.getInProcess() > this.outstandingLimit) {
					if (LOG.isDebugEnabled()) {
						LOG.debug("Throttling recv " + this.zkServer.getInProcess());
					}
					this.disableRecv();
					// following lines should not be needed since we are
					// already reading
					// } else {
					// enableRecv();
				}
			}
		}

	}

	/*
	 * (non-Javadoc)
	 *
	 * @see
	 * org.apache.zookeeper.server.ServerCnxnIface#process(org.apache.zookeeper.
	 * proto.WatcherEvent)
	 */
	@Override
	synchronized public void process(WatchedEvent event) {
		ReplyHeader h = new ReplyHeader(-1, -1L, 0);
		if (LOG.isTraceEnabled()) {
			ZooTrace.logTraceMessage(LOG, ZooTrace.EVENT_DELIVERY_TRACE_MASK,
					"Deliver event " + event + " to 0x" + Long.toHexString(this.sessionId) + " through " + this);
		}

		// Convert WatchedEvent to a type that can be sent over the wire
		WatcherEvent e = event.getWrapper();

		this.sendResponse(h, e, "notification");
	}

	private void readConnectRequest() throws IOException, InterruptedException {
		if (this.zkServer == null) {
			throw new IOException("ZooKeeperServer not running");
		}
		this.zkServer.processConnectRequest(this, this.incomingBuffer);
		this.initialized = true;
	}

	/**
	 * Reads the first 4 bytes of lenBuffer, which could be true length or four
	 * letter word.
	 *
	 * @param k selection key
	 * @return true if length read, otw false (wasn't really the length)
	 * @throws IOException if buffer size exceeds maxBuffer size
	 */
	private boolean readLength(SelectionKey k) throws IOException {
		// Read the length, now get the buffer
		int len = this.lenBuffer.getInt();
		if (!this.initialized && this.checkFourLetterWord(this.sk, len)) {
			return false;
		}
		if (len < 0 || len > BinaryInputArchive.maxBuffer) {
			throw new IOException("Len error " + len);
		}
		if (this.zkServer == null) {
			throw new IOException("ZooKeeperServer not running");
		}
		this.incomingBuffer = ByteBuffer.allocate(len);
		return true;
	}

	/** Read the request payload (everything following the length prefix) */
	private void readPayload() throws IOException, InterruptedException {
		if (this.incomingBuffer.remaining() != 0) { // have we read length bytes?
			int rc = this.sock.read(this.incomingBuffer); // sock is non-blocking, so ok
			if (rc < 0) {
				throw new EndOfStreamException("Unable to read additional data from client sessionid 0x"
						+ Long.toHexString(this.sessionId) + ", likely client has closed socket");
			}
		}

		if (this.incomingBuffer.remaining() == 0) { // have we read length bytes?
			this.packetReceived();
			this.incomingBuffer.flip();
			if (!this.initialized) {
				this.readConnectRequest();
			} else {
				this.readRequest();
			}
			this.lenBuffer.clear();
			this.incomingBuffer = this.lenBuffer;
		}
	}

	private void readRequest() throws IOException {
		this.zkServer.processPacket(this, this.incomingBuffer);
	}

	@Override
	public void sendBuffer(ByteBuffer bb) {
		try {
			if (bb != ServerCnxnFactory.closeConn) {
				// We check if write interest here because if it is NOT set,
				// nothing is queued, so we can try to send the buffer right
				// away without waking up the selector
				if ((this.sk.interestOps() & SelectionKey.OP_WRITE) == 0) {
					try {
						this.sock.write(bb);
					} catch (IOException e) {
						// we are just doing best effort right now
					}
				}
				// if there is nothing left to send, we are done
				if (bb.remaining() == 0) {
					this.packetSent();
					return;
				}
			}

			synchronized (this.factory) {
				this.sk.selector().wakeup();
				if (LOG.isTraceEnabled()) {
					LOG.trace("Add a buffer to outgoingBuffers, sk " + this.sk + " is valid: " + this.sk.isValid());
				}
				this.outgoingBuffers.add(bb);
				if (this.sk.isValid()) {
					this.sk.interestOps(this.sk.interestOps() | SelectionKey.OP_WRITE);
				}
			}

		} catch (Exception e) {
			LOG.error("Unexpected Exception: ", e);
		}
	}

	/**
	 * send buffer without using the asynchronous calls to selector and then close
	 * the socket
	 * 
	 * @param bb
	 */
	void sendBufferSync(ByteBuffer bb) {
		try {
			/*
			 * configure socket to be blocking so that we dont have to do write in a tight
			 * while loop
			 */
			this.sock.configureBlocking(true);
			if (bb != ServerCnxnFactory.closeConn) {
				if (this.sock != null) {
					this.sock.write(bb);
				}
				this.packetSent();
			}
		} catch (IOException ie) {
			LOG.error("Error sending data synchronously ", ie);
		}
	}

	/*
	 * Send close connection packet to the client, doIO will eventually close the
	 * underlying machinery (like socket, selectorkey, etc...)
	 */
	@Override
	public void sendCloseSession() {
		this.sendBuffer(ServerCnxnFactory.closeConn);
	}

	/*
	 * (non-Javadoc)
	 *
	 * @see
	 * org.apache.zookeeper.server.ServerCnxnIface#sendResponse(org.apache.zookeeper
	 * .proto.ReplyHeader, org.apache.jute.Record, java.lang.String)
	 */
	@Override
	synchronized public void sendResponse(ReplyHeader h, Record r, String tag) {
		try {
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			// Make space for length
			BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
			try {
				baos.write(fourBytes);
				bos.writeRecord(h, "header");
				if (r != null) {
					bos.writeRecord(r, tag);
				}
				baos.close();
			} catch (IOException e) {
				LOG.error("Error serializing response");
			}
			byte b[] = baos.toByteArray();
			ByteBuffer bb = ByteBuffer.wrap(b);
			bb.putInt(b.length - 4).rewind();
			this.sendBuffer(bb);
			if (h.getXid() > 0) {
				synchronized (this) {
					this.outstandingRequests--;
				}
				// check throttling
				synchronized (this.factory) {
					if (this.zkServer.getInProcess() < this.outstandingLimit || this.outstandingRequests < 1) {
						this.sk.selector().wakeup();
						this.enableRecv();
					}
				}
			}
		} catch (Exception e) {
			LOG.warn("Unexpected exception. Destruction averted.", e);
		}
	}

	@Override
	protected ServerStats serverStats() {
		if (this.zkServer == null) {
			return null;
		}
		return this.zkServer.serverStats();
	}

	@Override
	public void setSessionId(long sessionId) {
		this.sessionId = sessionId;
	}

	@Override
	public void setSessionTimeout(int sessionTimeout) {
		this.sessionTimeout = sessionTimeout;
	}

	@Override
	public String toString() {
		return "NIOServerCnxn object with sock = " + this.sock + " and sk = " + this.sk;
	}

}
